package cn.doitedu.udfs;

import cn.doitedu.pojo.LogBean;
import cn.doitedu.pojo.MatchResult;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.sql.Driver;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Map;

public class MatchFunction extends KeyedProcessFunction<String, LogBean, MatchResult> {

    private Connection hbaseConn;
    private java.sql.Connection clickHouseConn;

    @Override
    public void open(Configuration parameters) throws Exception {
        //在TaskManager初始化subtask时，获取全局参数
        ParameterTool parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();

        //初始化连接
        //1.创建Hbase的连接
        org.apache.hadoop.conf.Configuration hbaseConf = HBaseConfiguration.create();
        hbaseConf.set("hbase.zookeeper.quorum", parameterTool.getRequired("hbase.zookeeper.quorum"));
        hbaseConf.setInt("hbase.zookeeper.property.clientPort", parameterTool.getInt("hbase.zookeeper.property.clientPort", 2181));
        hbaseConn = ConnectionFactory.createConnection(hbaseConf);
        clickHouseConn = DriverManager.getConnection(parameterTool.getRequired("clickhouse.url"));
    }

    @Override
    public void processElement(LogBean bean, Context ctx, Collector<MatchResult> out) throws Exception {

        String deviceId = bean.getDeviceId();

        //2.匹配触发事件和行为属性
        //目前规则写死在程序中了，不灵活，以后使用广播状态，动态增加、修改、删除规则
        String eventId = bean.getEventId();
        Map<String, String> properties = bean.getProperties();
        if ("G".equals(eventId) && "v1".equals(properties.get("p4")) && "v1".equals(properties.get("p9"))) {

            //3.匹配用户画像(查询Hbase)
            Table table = hbaseConn.getTable(TableName.valueOf("eagle_profile"));
            //产线画像是否匹配
            Get get = new Get(Bytes.toBytes(deviceId));
            //添加过滤条件
            get.addColumn(Bytes.toBytes("f"), Bytes.toBytes("tag1"));
            get.addColumn(Bytes.toBytes("f"), Bytes.toBytes("tag3"));
            Result result = table.get(get);
            if (result != null) {
               String tag1Val = new String(result.getValue(Bytes.toBytes("f"), Bytes.toBytes("tag1")));
               String tag3Val = new String(result.getValue(Bytes.toBytes("f"), Bytes.toBytes("tag3")));

               if ("v2".equals(tag1Val) && "v2".equals(tag3Val)) {

                   //匹配历史行为数据（行为次数和行为序列）
                   //查询ClickHouse

                   //clickhouse可以使用jdbc进行查询
                   //select count(*) from eagle_detail where deviceId = '000005' and timeStamp >= 1649433600000 and timeStamp <= 1649606399000 and eventId = 'G' and properties['p4'] = 'v1';
                   PreparedStatement preparedStatement = clickHouseConn.prepareStatement("select count(*) from eagle_detail where deviceId = ? and timeStamp >= ? and timeStamp <= ? and eventId = 'G' and properties['p4'] = 'v1'");
                   preparedStatement.setString(1, deviceId);
                   preparedStatement.setLong(2, 1649433600000L);
                   preparedStatement.setLong(3, 1649606399000L);
                   ResultSet resultSet = preparedStatement.executeQuery();
                   long count = 0;
                   if(resultSet.next()) {
                       count = resultSet.getLong(1);
                   }
                   //判断次数是否满足
                   if(count >= 2) {
                       //在ip行为序列
                       //输出结果
                       //匹配行为序列
                       preparedStatement = clickHouseConn.prepareStatement("SELECT sequenceMatch('(?1)(?2)(?3)')(toDateTime(`timeStamp`), eventId = 'A',eventId = 'H',eventId = 'A') AS is_match FROM eagle_detail where deviceId = ? GROUP BY deviceId");
                       preparedStatement.setString(1, deviceId);
                       resultSet = preparedStatement.executeQuery();
                       int isMatch = 0;
                       if (resultSet.next()) {
                           isMatch = resultSet.getInt(1);
                       }
                       if (isMatch > 0) {
                           out.collect(new MatchResult(deviceId, bean.getTimeStamp(), "rule001", System.currentTimeMillis()));
                       }
                   }
                   resultSet.close();
                   preparedStatement.close();
               }
            }
            table.close();
        }

    }


    @Override
    public void close() throws Exception {
        //释放资源
        hbaseConn.close();
        clickHouseConn.close();

    }
}
